/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.indexer.hyracks;

import java.nio.ByteBuffer;
import java.util.logging.Level;
import java.util.logging.Logger;

import edu.uci.ics.hyracks.api.comm.IFrameWriter;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.dataflow.common.comm.io.ArrayTupleBuilder;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameDeserializer;
import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAppender;
import edu.uci.ics.hyracks.dataflow.common.comm.util.FrameUtils;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.BooleanSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.indexer.api.IPosting;
import edu.ucsd.db.bassarisk.indexer.api.IPosting.IPositionedPosting;
import edu.ucsd.db.bassarisk.indexer.api.IPosting.ISequencedPosting;
import edu.ucsd.db.bassarisk.indexer.api.IPostingHandler;
import edu.ucsd.db.bassarisk.indexer.impl.PostingFactory;

public class HyracksPostingSerialization {

  private static abstract class AbstractPostingTupleWriter<T extends IPosting>
      implements IPostingTupleWriter<T> {
    private final Logger LOG = Logger.getLogger(this.getClass().getName());
    private final int OUTPUT_TUPLE_SIZE = 4;

    private final ByteBuffer outputFrame;
    private final FrameTupleAppender appender;

    private final ArrayTupleBuilder tupleBuilder;

    private IFrameWriter frameWriter;

    private AbstractPostingTupleWriter(final ByteBuffer outputFrame,
                                       final FrameTupleAppender tupleAppender) {
      this.outputFrame = outputFrame;
      this.appender = tupleAppender;
      this.appender.reset(outputFrame, true);
      this.tupleBuilder = new ArrayTupleBuilder(this.OUTPUT_TUPLE_SIZE);
    }

    protected abstract void buildTuple(T posting, ArrayTupleBuilder tupleBuilder)
        throws HyracksDataException;

    @Override
    public void close() {
      LOG.info("Closing posting tuple writer.");
      // There may be postings left in the buffer. Flush 'em out.
      if (this.appender.getTupleCount() > 0) {
        this.flushOutput();
      }
    }

    private void flushOutput() {
      LOG.info(String.format("Flushing output frame with %s tuples.", 
                             this.appender.getTupleCount()));
      try {
        FrameUtils.flushFrame(this.outputFrame, this.frameWriter);
      } catch (HyracksDataException e) {
        this.LOG.log(Level.SEVERE,
                     "Hyrack frame flush failed. Frame of postings lost.", e);
      }
    }

    @Override
    public void post(final T posting) {
      // Use the tuple builder to assemble a new posting tuple
      try {
        this.tupleBuilder.reset();
        this.buildTuple(posting, this.tupleBuilder);
      } catch (HyracksDataException e) {
        this.LOG.log(Level.SEVERE,
                     "Hyrack tuple construction failed. Posting tuple lost.", e);
      }

      // Append the new posting tuple to the output buffer
      if (!this.appender.append(this.tupleBuilder.getFieldEndOffsets(),
                                this.tupleBuilder.getByteArray(), 0,
                                this.tupleBuilder.getSize())) {
        // Append probably failed because the buffer is too full. Flush
        // buffer and try again.
        this.flushOutput();
        this.appender.reset(this.outputFrame, true);
        if (!this.appender.append(this.tupleBuilder.getFieldEndOffsets(),
                                  this.tupleBuilder.getByteArray(), 0,
                                  this.tupleBuilder.getSize())) {
          String failureMessage =
              "Failure attempting to append data tuple to frame. Append fails after frame flush.";
          this.LOG.severe(failureMessage);
          throw new IllegalStateException(failureMessage);
        }
      }
    }

    @Override
    public void setFrameWriter(final IFrameWriter newWriter) {
      this.frameWriter = newWriter;
    }
  }

  public static interface IPostingTupleReader<T extends IPosting> {
    void nextFrame(ByteBuffer buffer);
  }

  public static interface IPostingTupleWriter<T extends IPosting> extends
      IPostingHandler<T> {
    void close();
    void setFrameWriter(final IFrameWriter newWriter);
  }

  public static
      IPostingTupleReader<IPositionedPosting>
      createPositionedPostingTupleReader(final int frameSize,
                                         final IPostingHandler<IPositionedPosting> postingSink,
                                         final PostingFactory postingFactory) {
    final FrameDeserializer fd =
        new FrameDeserializer(frameSize, positionedPostingsDescriptor());

    return new IPostingTupleReader<IPositionedPosting>() {
      @Override
      public void nextFrame(final ByteBuffer buffer) {
        fd.reset(buffer);
        while (!fd.done()) {
          try {
            Object[] fields = fd.deserializeRecord();
            IPositionedPosting posting =
                postingFactory.posting((String) fields[0], (String) fields[1],
                                       (IValueReference<?>) fields[2],
                                       (Integer) fields[3]);
            postingSink.post(posting);
          } catch (HyracksDataException e) {
            throw new RuntimeException(e);
          }
        }
      }
    };
  }

  public static IPostingTupleWriter<IPositionedPosting>
      createPositionedPostingTupleWriter(final ByteBuffer outputFrame,
                                         final FrameTupleAppender tupleAppender) {
    return new AbstractPostingTupleWriter<IPositionedPosting>(outputFrame,
                                                              tupleAppender) {
      @Override
      protected void buildTuple(final IPositionedPosting posting,
                                final ArrayTupleBuilder tupleBuilder)
          throws HyracksDataException {
        tupleBuilder.addField(UTF8StringSerializerDeserializer.INSTANCE,
                              posting.getGroupKey());
        tupleBuilder.addField(UTF8StringSerializerDeserializer.INSTANCE,
                              posting.getTokenKey());
        tupleBuilder.addField(MixedTypeSerailizerDeserializer.INSTANCE,
                              posting.getValue());
        tupleBuilder.addField(IntegerSerializerDeserializer.INSTANCE,
                              (int) posting.getPosition());
        // FIXME (nbales) -- Add to Hyracks support for
        // LongSerializerDeserializer
      }

    };
  }

  public static
      IPostingTupleReader<ISequencedPosting>
      createSequencedPostingTupleReader(final int frameSize,
                                        final IPostingHandler<ISequencedPosting> postingSink,
                                        final PostingFactory postingFactory) {
    final FrameDeserializer fd =
        new FrameDeserializer(frameSize, sequencedPostingsDescriptor());

    return new IPostingTupleReader<ISequencedPosting>() {
      @Override
      public void nextFrame(final ByteBuffer buffer) {
        fd.reset(buffer);
        while (!fd.done()) {
          try {
            Object[] fields = fd.deserializeRecord();
            ISequencedPosting posting =
                postingFactory.posting((String) fields[0], (String) fields[1],
                                       (IValueReference<?>) fields[2],
                                       (Boolean) fields[3]);
            postingSink.post(posting);
          } catch (HyracksDataException e) {
            throw new RuntimeException(e);
          }
        }
      }
    };
  };

  public static IPostingTupleWriter<ISequencedPosting>
      createSequencedPostingTupleWriter(final ByteBuffer outputFrame,
                                        final FrameTupleAppender tupleAppender) {
    return new AbstractPostingTupleWriter<ISequencedPosting>(outputFrame,
                                                             tupleAppender) {
      @Override
      protected void buildTuple(final ISequencedPosting posting,
                                final ArrayTupleBuilder tupleBuilder)
          throws HyracksDataException {
        tupleBuilder.addField(UTF8StringSerializerDeserializer.INSTANCE,
                              posting.getGroupKey());
        tupleBuilder.addField(UTF8StringSerializerDeserializer.INSTANCE,
                              posting.getTokenKey());
        tupleBuilder.addField(MixedTypeSerailizerDeserializer.INSTANCE,
                              posting.getValue());
        tupleBuilder.addField(BooleanSerializerDeserializer.INSTANCE,
                              posting.consumesSpace());
      }

    };
  }

  public static RecordDescriptor positionedPostingsDescriptor() {
    return new RecordDescriptor(
                                new ISerializerDeserializer[] {
                                                               UTF8StringSerializerDeserializer.INSTANCE,
                                                               UTF8StringSerializerDeserializer.INSTANCE,
                                                               MixedTypeSerailizerDeserializer.INSTANCE,
                                                               IntegerSerializerDeserializer.INSTANCE });
  }

  public static RecordDescriptor sequencedPostingsDescriptor() {
    return new RecordDescriptor(
                                new ISerializerDeserializer[] {
                                                               UTF8StringSerializerDeserializer.INSTANCE,
                                                               UTF8StringSerializerDeserializer.INSTANCE,
                                                               MixedTypeSerailizerDeserializer.INSTANCE,
                                                               BooleanSerializerDeserializer.INSTANCE });
  }
}
